package spring.cloud.tasks.common.zookeeper.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.transaction.CuratorTransactionFinal;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.zookeeper.CreateMode;
import spring.cloud.tasks.common.zookeeper.TransactionalZooKeeper;

import java.nio.charset.StandardCharsets;
import java.util.Collection;

@Slf4j
class TransactionalZooKeeperImpl implements TransactionalZooKeeper {

    private final CuratorFramework curatorFramework;
    private CuratorTransactionFinal curatorTransactionFinal;

    public TransactionalZooKeeperImpl(CuratorFramework curatorFramework) {
        this.curatorFramework = curatorFramework;
        try {
            curatorTransactionFinal = curatorFramework.inTransaction().check().forPath("/").and();
        } catch (final Exception ex) {
            throw new RuntimeException(ex);
        }
    }

    private boolean checkExists(String path) throws Exception {
        return curatorFramework.checkExists().forPath(path) != null;
    }

    private TransactionalZooKeeperImpl create(String path, byte[] data) throws Exception {
        curatorTransactionFinal = curatorTransactionFinal.create().withMode(CreateMode.PERSISTENT).forPath(path, data).and();
        return this;
    }

    private byte[] getData(String path) throws Exception {
        return curatorFramework.getData().forPath(path);
    }

    private boolean bytesEquals(byte[] a, byte[] b) {
        if (a == null || b == null) {
            return (a == null && b == null);
        }
        if (a.length != b.length) {
            return false;
        }
        for (int i = 0, size = a.length; i < size; i++) {
            if (a[i] != b[i]) {
                return false;
            }
        }
        return true;
    }

    @Override
    public TransactionalZooKeeperImpl replace(String path, Object value) throws Exception {
        if (value == null) {
            log.warn("node value is null, won't replace, node: {}", path);
            return this;
        }
        byte[] data = value.toString().getBytes(StandardCharsets.UTF_8);
        curatorTransactionFinal = curatorTransactionFinal.setData().forPath(path, data).and();
        return this;
    }

    @Override
    public TransactionalZooKeeperImpl replaceIfChanged(String path, Object value) throws Exception {
        if (value == null) {
            log.warn("node value is null, won't replaceIfChanged, node: {}", path);
            return this;
        }
        byte[] newData = value.toString().getBytes(StandardCharsets.UTF_8);
        if (this.checkExists(path)) {
            byte[] oldData = this.getData(path);
            if (!bytesEquals(newData, oldData)) {
                curatorTransactionFinal = curatorTransactionFinal.check().forPath(path).and().setData().forPath(path, newData).and();
            }
        } else {
            this.create(path, newData);
        }
        return this;
    }

    @Override
    public TransactionalZooKeeper create(String path, Object value) throws Exception {
        if (value == null) {
            log.warn("node value is null, won't create, node: {}", path);
            return this;
        }
        byte[] data = value.toString().getBytes(StandardCharsets.UTF_8);
        curatorTransactionFinal = curatorTransactionFinal.create().withMode(CreateMode.PERSISTENT).forPath(path, data).and();
        return this;
    }

    @Override
    public TransactionalZooKeeper delete(String path) throws Exception {
        curatorTransactionFinal = curatorTransactionFinal.delete().forPath(path).and();
        return this;
    }

    @Override
    public Collection<CuratorTransactionResult> commit() throws Exception {
        return curatorTransactionFinal.commit();
    }
}
